package com.atguigu.gmall.realtime.util

import java.util
import java.util.Properties

import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.serializer.SerializeConfig
import org.apache.http.HttpHost
import org.elasticsearch.action.bulk.BulkRequest
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.action.search.{SearchRequest, SearchResponse}
import org.elasticsearch.client.{RequestOptions, RestClient, RestClientBuilder, RestHighLevelClient}
import org.elasticsearch.common.xcontent.XContentType
import org.elasticsearch.index.query.{BoolQueryBuilder, MatchQueryBuilder, RangeQueryBuilder}
import org.elasticsearch.search.SearchHit
import org.elasticsearch.search.aggregations.AggregationBuilders
import org.elasticsearch.search.aggregations.bucket.terms.{ParsedLongTerms, ParsedStringTerms, Terms, TermsAggregationBuilder}
import org.elasticsearch.search.builder.SearchSourceBuilder
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder

object MyEsUtil {


  private val properties: Properties =  MyPropertiesUtil.load("config.properties")
  val esHost = properties.getProperty("es.host")
  val esPort = properties.getProperty("es.port")
  private var  restHighLevelClient:RestHighLevelClient=getClientNew

  def getClientNew:RestHighLevelClient={
    if(restHighLevelClient==null){
      val restClientBuilder: RestClientBuilder = RestClient.builder(new HttpHost(esHost,esPort.toInt))
      restHighLevelClient= new RestHighLevelClient(restClientBuilder)
    }
    restHighLevelClient
  }

  //单条写入
  def   saveDoc(): Unit ={
    val movieJson="{\n  \"id\":\"0104\",\n  \"movie_name\":\"天龙八部888888888\"\n}"
    val indexRequest = new IndexRequest("movie_a0722_bbbbb").`type`("_doc").source(movieJson,XContentType.JSON)
    restHighLevelClient.index( indexRequest, RequestOptions.DEFAULT)

  }

  // bulk = batch  //批量写入
  def saveBulk( sourceList: List[(String,AnyRef)],indexName:String): Unit ={

    val bulkRequest = new BulkRequest(indexName)

    for ((id,source) <- sourceList ) {
      val json: String = JSON.toJSONString(source,new SerializeConfig(true))
      val indexRequest = new IndexRequest().source(json,XContentType.JSON).id(id)
      bulkRequest.add(indexRequest)
    }

    restHighLevelClient.bulk(bulkRequest,RequestOptions.DEFAULT)
    println(s"已保存${sourceList.size}条数据！")
  }

  def main(args: Array[String]): Unit = {
    //saveDoc()
  //  val movieList = List(Movie("1010","笑傲江湖"),Movie("1012","神雕侠侣"))
  //  saveBulk(movieList,"movie_a0722_bbbbb")

    //测试查询
   // search()
    //测试查询聚合
    aggs()
    restHighLevelClient.close() //持续运行的程序不要关



  }

  def  search(): Unit ={
    val searchSourceBuilder = new SearchSourceBuilder()
    //查询
    val boolQueryBuilder = new BoolQueryBuilder()
    boolQueryBuilder.filter(new RangeQueryBuilder("doubanScore").gte(7).lte(10))
    boolQueryBuilder.must(new MatchQueryBuilder("name","红海"))

    searchSourceBuilder.query(boolQueryBuilder)

    //高亮
    val highlightBuilder: HighlightBuilder = new HighlightBuilder().field("name")
    searchSourceBuilder.highlighter(highlightBuilder)

    //分页部分
    searchSourceBuilder.from(0)
    searchSourceBuilder.size(20)


    val searchRequest = new SearchRequest("movie_chn0722").source(searchSourceBuilder)
    val searchResponse: SearchResponse = restHighLevelClient.search(searchRequest,RequestOptions.DEFAULT)

    val total: Long = searchResponse.getHits.getTotalHits.value
    println(total)
    val hits: Array[SearchHit] = searchResponse.getHits.getHits
    for (hit <- hits ) {
      println(hit.getSourceAsString)
    }

  }

  def aggs(): Unit ={
    val searchSourceBuilder = new SearchSourceBuilder()
    searchSourceBuilder.size(0)

    val aggregationBuilder: TermsAggregationBuilder = AggregationBuilders.terms("groupby_actor").field("actorList.name").size(100)
    searchSourceBuilder.aggregation(aggregationBuilder)

    val searchRequest = new SearchRequest("movie_chn0722").source(searchSourceBuilder)
    val searchResponse: SearchResponse = restHighLevelClient.search(searchRequest,RequestOptions.DEFAULT)

    val groupByActorTerms: ParsedStringTerms = searchResponse.getAggregations.get[ParsedStringTerms]("groupby_actor")
    val buckets: util.List[_ <: Terms.Bucket] = groupByActorTerms.getBuckets
    import  scala.collection.JavaConverters._
    for (bucket <- buckets.asScala) {
      println(bucket.getKey)
      println(bucket.getDocCount)
    }
  }

  case class Movie(id:String ,movie_name:String)





}
